package com.shujia.state

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.JavaConversions._

object Demo04ListState {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 每1000ms做一次checkpoint
    env.enableCheckpointing(1000)
    // 高级选项(可选)
    // 设置CheckPoint的模式为EXACTLY_ONCE精确一次/完全一次(默认)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 设置两个CheckPoint任务之间的时间间隔
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // 设置CheckPoint的超时时间
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    // 设置同一时刻最多能有多少个CheckPoint任务
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    // 设置在任务取消时不清理CheckPoint保存的状态
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    // 设置CheckPoint目录 使用文件系统作为状态后端（保存状态的地方）
    env.setStateBackend(new FsStateBackend("hdfs://master:9000/flink/wc/checkpoint"))

    val stuDS: DataStream[String] = env.socketTextStream("master", 8888)

    val clazzAgeDS: DataStream[(String, Int)] = stuDS
      .map(stu => {
        val stuArr: Array[String] = stu.split(",")
        (stuArr(4), stuArr(2).toInt)
      })
    /**
     * 每个班级的平均年龄 并保存状态
     */

    clazzAgeDS
      .keyBy(_._1)
      .process(new KeyedProcessFunction[String, (String, Int), (String, Double)] {

        var ageListState: ListState[Int] = _

        override def open(parameters: Configuration): Unit = {
          // 获取Flink运行时环境
          val ctx: RuntimeContext = getRuntimeContext
          // 创建ListState描述
          val ageListStateDesc: ListStateDescriptor[Int] = new ListStateDescriptor[Int]("ages", classOf[Int])
          // 获取一个ListsState
          ageListState = ctx.getListState(ageListStateDesc)

        }

        override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Double)]#Context, out: Collector[(String, Double)]): Unit = {
          val (clazz, age): (String, Int) = value
          // 从每个班级（相当于每一个Key）过来的数据 将age提取出来 放入ListState中
          ageListState.add(age)
          // 取出一个班级所有的年龄 并转成scala中的List集合
          val ages: List[Int] = ageListState.get().iterator().toList
          // 计算班级年龄的平均值
          val avgAge: Double = ages.sum.toDouble / ages.size
          out.collect((clazz, avgAge))
        }
      })
      .print()



    env.execute()
  }

}
